package chapter4.eg11;

import chapter4.AbstractProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Test;

public class SerializerTest extends AbstractProducer {

  @Test
  public void testSerializer() throws ExecutionException, InterruptedException {
    List<String> interceptors = new ArrayList<>();
    interceptors.add("chapter4.eg12.TimestampPrependerInterceptor");
    interceptors.add("chapter4.eg12.CounterInterceptor");
    properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "chapter4.eg11.UserSerializer");

    String topic = "test-topic";
    Producer<String, User> producer = new KafkaProducer<>(properties);
    User user = new User("XI", "HU", 33, "Beijing, China");
    ProducerRecord<String, User> record = new ProducerRecord<>(topic, user);
    RecordMetadata recordMetadata = producer.send(record).get();
    System.out.println(recordMetadata.toString());
    producer.close();
  }
}
